package com.quwan.chat.handle;


import com.alibaba.druid.support.json.JSONUtils;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.quwan.chat.module.DataContent;
import com.quwan.entity.ChatOfflineMsg;
import com.quwan.mapper.ChatOfflineMsgMapper;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


/**
 * @author quan
 * @date 2021-09-01 11:03
 */
@ChannelHandler.Sharable
@Component
public class ChatMsgHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private static final Logger log = LoggerFactory.getLogger(ChatMsgHandler.class);
    private static final ConcurrentHashMap<String, ChannelId> M_CHANNEL = new ConcurrentHashMap();
    public static final ChannelGroup cg = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private static final AttributeKey<String> CHANNEL_MEMBER_KEY = AttributeKey.valueOf("member_channel_id");
    @Autowired
    private ChatOfflineMsgMapper chatOfflineMsgMapper;

    /**
     * 聊天 消息读取处理
     * 接收消息体json格式{@link DataContent}
     * @param ctx
     * @param msg
     * @throws Exception
     */

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        if (msg == null || StringUtils.isBlank(msg.text())) return;
        DataContent dataContent = JSON.parseObject(msg.text(), DataContent.class);
        if (dataContent == null) return;
        Channel channel = ctx.channel();
        ChannelId channelId = channel.id();
        if ("0".equalsIgnoreCase(dataContent.getType())) {
            String connectMember = dataContent.getConnectId();
            //建立聊天会话
            if (StringUtils.isBlank(connectMember)) return;
            channel.attr(CHANNEL_MEMBER_KEY).set(connectMember);
            M_CHANNEL.put(connectMember, channelId);
            cg.add(channel);
            //查询用户离线消息
            QueryWrapper<ChatOfflineMsg> wrapper = new QueryWrapper<>(
                    new ChatOfflineMsg()
                            .setReceiverId(Long.parseLong(connectMember)));
            List<String> offlineMsgs = chatOfflineMsgMapper
                    .selectList(wrapper)
                    .stream()
                    .map(JSON::toJSONString)
                    .collect(Collectors.toList());

            if (CollectionUtils.isNotEmpty(offlineMsgs)) {
                channel.writeAndFlush(
                        new TextWebSocketFrame(offlineMsgs.toString()));
                chatOfflineMsgMapper.delete(wrapper);
            }
        } else if ("1".equalsIgnoreCase(dataContent.getType())) {
            //互动（聊天）
            String receiverId = dataContent.getReceiverId() == null ? "" : dataContent.getReceiverId();
            ChannelId receiverChannelId = M_CHANNEL.get(receiverId);
            Channel receiverChannel = null;
            if (receiverChannelId != null) receiverChannel = cg.find(receiverChannelId);
            if (receiverChannel != null && receiverChannel.isActive()
                    && receiverId.equalsIgnoreCase(receiverChannel.attr(CHANNEL_MEMBER_KEY).get())) {
                dataContent.setDate(new Date());
                //会员channel 存在
                receiverChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(dataContent)));
            } else {
                //channel 不存在  离线消息存储DB
                log.debug("member offline msg2DB");
                chatOfflineMsgMapper.insert(
                        new ChatOfflineMsg()
                                .setReceiverId(Long.parseLong(receiverId))
                                .setContent(dataContent.getContent())
                                .setSenderId(Long.parseLong(dataContent.getSenderId()))
                                .setSessionId(dataContent.getSessionId()));
            }

        }
    }


    /**
     * 断开事件处理
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        super.handlerRemoved(ctx);
        String key = ctx.channel().attr(CHANNEL_MEMBER_KEY).get();
       if (StringUtils.isNotBlank(key))M_CHANNEL.remove(key);
        log.debug("member断开连接:{}",key);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
        super.exceptionCaught(ctx, cause);
        log.error("chat channel exception:{}",cause.getMessage());
    }
}
